package com.young.springboot.kafka;

import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.apache.kafka.clients.consumer.ConsumerConfig;
import org.springframework.beans.factory.DisposableBean;
import org.springframework.kafka.config.ConcurrentKafkaListenerContainerFactory;
import org.springframework.kafka.config.KafkaListenerEndpoint;
import org.springframework.kafka.listener.ConcurrentMessageListenerContainer;
import org.springframework.kafka.support.TopicPartitionInitialOffset;
import org.springframework.util.StringUtils;

import java.util.Collection;
import java.util.concurrent.Executors;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.TimeUnit;

/**
 * 提供了一个自动重启功能的 Kafka 消息监听器容器工厂。
 * 如果容器意外停止，它会自动尝试重启容器，确保消息监听的高可用性。
 * 通过定时任务和线程同步机制实现了自动重启逻辑，并且可以通过 groupId 方法获取 Kafka 消费者组 ID。
 * @description：
 * @author: yangyahui01
 * @date: 12/5/24 2:52 PM
 */
public class ScheduledConcurrentKafkaListenerContainerFactory<K,V> extends ConcurrentKafkaListenerContainerFactory<K, V> implements DisposableBean {

    private ScheduledExecutorService executorService = Executors.newScheduledThreadPool(1);

    private Object lock = new Object();

    protected final Log logger = LogFactory.getLog(this.getClass()); // NOSONAR

    private volatile boolean closed = false;

    private boolean enableAutoRestart;

    public ScheduledConcurrentKafkaListenerContainerFactory(boolean enableAutoRestart) {
        this.enableAutoRestart = enableAutoRestart;
    }


    @Override
    protected ConcurrentMessageListenerContainer<K, V> createContainerInstance(KafkaListenerEndpoint endpoint) {
        ConcurrentMessageListenerContainer<K, V> container = super.createContainerInstance(endpoint);
        if (!enableAutoRestart || executorService.isShutdown()) {
            return container;
        }

        Collection<TopicPartitionInitialOffset> topicPartitions = endpoint.getTopicPartitions();
        //对于指定初始值的，不能开启schedule
        if (!topicPartitions.isEmpty()) {
            return container;
        }

        executorService.scheduleAtFixedRate(new Runnable() {
            @Override
            public void run() {
                synchronized (lock) {
                    //如果正常运行，则不操作
                    if (container.isRunning()) {
                        return;
                    }
                    //如果当前factory已经关闭，则补充stop操作
                    if (closed) {
                        container.stop();
                        return;
                    }
                    //factory没有关闭，且container被关闭时
                    String cid = "{groupId:" + groupId(endpoint) + ",topics:" + endpoint.getTopics() + "}";
                    try {
                        logger.info("Restarting container," + cid);
                        container.start();
                        logger.info("Restart container successfully," + cid);
                    } catch (Exception e) {
                        logger.error("Restart container error," + cid,e);
                    }
                }
            }
        },60,180, TimeUnit.SECONDS);

        return container;
    }


    //私有，不覆盖
    private String groupId(KafkaListenerEndpoint endpoint) {
        String groupId = endpoint.getGroupId();
        if (StringUtils.hasText(groupId)) {
            return groupId;
        }

        if (getConsumerFactory() != null) {
            groupId = (String)getConsumerFactory().getConfigurationProperties()
                    .get(ConsumerConfig.GROUP_ID_CONFIG);
            if (StringUtils.hasText(groupId)) {
                return groupId;
            }
        }
        return getContainerProperties().getGroupId();
    }

    @Override
    public void destroy() throws Exception {
        closed = true;
        executorService.shutdownNow();
    }
}
